import {
  App,
  IMidwayApplication,
  Provide,
  Scope,
  ScopeEnum,
} from '@midwayjs/core';
import { Inject } from '@midwayjs/decorator';
import { BaseService, CoolCommException } from '@cool-midway/core';
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Repository } from 'typeorm';
import { TaskTopicEntity } from '../entity/topic';
import { TaskInitService } from './init';
import { KeysMap, Queue, Worker } from 'bullmq';
import axios from 'axios';
import { TaskJobService } from './job';
import { TaskJobEntity } from '../entity/job';
import * as mqtt from 'mqtt';
import { TaskRecordEntity } from '../entity/record';
import { TaskRecordService } from './record';

/**
 * 话题
 */
@Scope(ScopeEnum.Singleton, { allowDowngrade: true })
@Provide()
export class TaskTopicService extends BaseService {
  @InjectEntityModel(TaskTopicEntity)
  taskTopicEntity: Repository<TaskTopicEntity>;

  @InjectEntityModel(TaskJobEntity)
  taskJobEntity: Repository<TaskJobEntity>;

  @InjectEntityModel(TaskRecordEntity)
  taskRecordEntity: Repository<TaskRecordEntity>;

  @Inject()
  taskInitService: TaskInitService;

  @Inject()
  taskJobService: TaskJobService;

  @Inject()
  taskRecordService: TaskRecordService;

  @App()
  app: IMidwayApplication;

  /**
   * 修改之前
   * @param data
   * @param type
   */
  async modifyBefore(data: any, type: 'delete' | 'update' | 'add') {
    if (type == 'add' || type == 'update') {
      // 判断connectOptions 如果不为空，需要判断是否是json格式的字符串
      if (data.connectOptions) {
        try {
          let result = JSON.parse(data.connectOptions);
          if (typeof result !== 'object' || result === null) {
            throw new Error();
          }
        } catch (error) {
          throw new CoolCommException('connectOptions需要是json格式的字符串');
        }
      } else {
        data.connectOptions = null;
      }
    }
  }

  /**
   * 新增
   * @param param
   * @returns
   */
  async add(param: TaskTopicEntity) {
    const check = await this.taskTopicEntity.findOneBy({ name: param.name });
    if (check) {
      throw new CoolCommException('名称已存在');
    }
    const result = await super.add(param);
    await this.create(param.id);
    return result;
  }

  /**
   * 更新
   * @param param
   */
  async update(param: TaskTopicEntity) {
    const old = await this.taskTopicEntity.findOneBy({ id: param.id });
    if (param.name && old.name != param.name) {
      const check = await this.taskTopicEntity.findOneBy({ name: param.name });
      if (check && check.id != param.id) {
        throw new CoolCommException('名称已存在');
      }
      await this.destroy(old.id);
      await super.update(param);
      await this.create(param.id);
      return;
    }
    await super.update(param);
    if (param.status == 0) {
      await this.destroy(param.id);
    }
    if (old.status == 0 && param.status == 1) {
      await this.create(param.id);
    }
  }

  /**
   * 删除
   * @param ids
   */
  async delete(ids: any): Promise<void> {
    for (const id of ids) {
      await this.destroy(id);
    }
    await super.delete(ids);
  }

  /**
   * 销毁队列
   * @param id
   */
  async destroy(id: number) {
    const topic = await this.taskTopicEntity.findOneBy({ id });
    const name = `${topic.appId}@${topic.name}`;
    const queue = await this.taskInitService.getQueue(topic.appId, topic.name);
    const worker = await this.taskInitService.getWorker(
      topic.appId,
      topic.name
    );
    if (!queue || !worker) {
      return;
    }
    await queue?.obliterate({ force: true });
    const keys: KeysMap = queue.keys;
    for (const key in keys) {
      await this.taskInitService.redis.del(keys[key]);
    }
    worker.close();
    queue.close();
    delete this.app['QUEUES'][name];
    delete this.app['WORKERS'][name];
  }

  /**
   * 创建
   * @param id
   */
  async create(id: number) {
    const topic = await this.taskTopicEntity.findOneBy({ id });
    const name = `${topic.appId}@${topic.name}`;
    const connection = this.taskInitService.redis;
    // 队列配置
    const option = {
      connection,
      defaultJobOptions: {
        removeOnComplete: true,
        removeOnFail: true,
        attempts: 5,
        backoff: {
          type: 'fixed',
          delay: 10000,
        },
      },
    };
    const queue = new Queue(name, option);
    const worker = new Worker(
      name,
      async job => {
        await this.handleJob(job);
      },
      { connection }
    );
    this.app['QUEUES'][name] = queue;
    this.app['WORKERS'][name] = worker;
    return queue;
  }

  /**
   * 处理任务
   * @param job
   */
  async handleJob(job) {
    const { topic, jobId, appId } = job.data;
    const topicInfo = await this.taskTopicEntity.findOneBy({
      appId,
      name: topic,
      status: 1,
    });
    if (!topicInfo) return;
    const info = await this.taskJobEntity.findOneBy({
      topic,
      jobId,
      appId,
    });
    // 生成任务记录
    const recordId = await this.taskRecordService.save({
      jobId: info.id,
      appId: info.appId,
      topic: info.topic,
    });
    if (topicInfo.type == 0) {
      await this.handleHTTP(topicInfo, info, recordId);
    }
    if (topicInfo.type == 1) {
      await this.handleMQTT(topicInfo, info, recordId);
    }
    // 如果是重复任务，需要更新下次执行时间
    if (info.type == 0) {
      this.taskJobService.updateNextRunTime(info.id);
    }
    if (info.type == 1) {
      // 更新任务状态为已完成
      this.taskJobEntity.update(info.id, {
        status: 2,
      });
    }

    job.isCompleted();
  }

  /**
   * 处理HTTP回调
   * @param topic
   * @param info
   */
  async handleHTTP(
    topic: TaskTopicEntity,
    info: TaskJobEntity,
    recordId: number
  ) {
    const { HTTPUrl, connectOptions = {} } = topic;
    await axios({
      url: HTTPUrl,
      method: 'POST',
      timeout: 10 * 1000,
      data: {
        recordId,
        ...info.body,
      },
      ...connectOptions,
    }).catch(err => {
      this.taskRecordEntity.update(recordId, {
        status: 2,
        result: err.message,
        resultTime: new Date(),
      });
    });
  }

  /**
   * 处理MQTT回调
   * @param topic
   * @param info
   */
  async handleMQTT(
    topic: TaskTopicEntity,
    info: TaskJobEntity,
    recordId: number
  ) {
    const { MQTTUrl, connectOptions = {}, name } = topic;
    const client = mqtt.connect(MQTTUrl, {
      connectTimeout: 10 * 1000,
      ...connectOptions,
    });
    client.on('connect', () => {
      client.subscribe(name, err => {
        if (!err) {
          client.publish(
            name,
            JSON.stringify({
              recordId,
              ...info.body,
            })
          );
          client.end();
        } else {
          this.taskRecordEntity.update(recordId, {
            status: 2,
            result: err.message,
            resultTime: new Date(),
          });
        }
      });
    });
  }
}
